package sql;

import java.util.Base64;

public class Bse64Test {

    public static void main(String[] args) {

        String sql = "create table o_ops_miner_quality_monitor (\n" +
                "    `stime`\ttimestamp(3),\n" +
                "    `shostname`\tstring,\n" +
                "    `hosten`\tstring,\n" +
                "    `metric`\tstring,\n" +
                "    `tags`\tstring,\n" +
                "    `val`\tdouble,\n" +
                "    `source`\tstring,\n" +
                "    proctime as proctime(),\n" +
                "    watermark for `stime` as `stime` - interval '60' second\n" +
                ") with (\n" +
                "    'properties.bootstrap.servers' = 'bdyun-suzhou-tw04a0066.su.baidu.internal:6667,bdyun-suzhou-tw04b0077.su.baidu.internal:6667,bdyun-suzhou-tw04d0639.su.baidu.internal:6667',\n" +
                "    'connector' = 'kafka',\n" +
                "    'format' = 'avro-event',\n" +
                "    'topic' = 'o_ops_miner_quality_monitor',\n" +
                "    'properties.group.id' = 'beta-k8s-lichen-o_ops_miner_quality_monitor-test',\n" +
                "    'properties.security.protocol' = 'SASL_PLAINTEXT',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'properties.sasl.kerberos.service.name' = 'kafka'\n" +
                ");\n" +
                "\n" +
                "create table d_idc_details (\n" +
                "    `idc` string,\n" +
                "    `switch_name`\tstring,\n" +
                "    `name`\tstring,\n" +
                "    `isp`\tint,\n" +
                "    `port`\tint,\n" +
                "    `status`\tint,\n" +
                "    `collect`\tint,\n" +
                "    `alarm`\tint,\n" +
                "    `bandwith` string,\n" +
                "    `idc_logic`\tstring,\n" +
                "    `bandwith_link`\tdouble\n" +
                ") with (\n" +
                "    'connector' = 'tidb',\n" +
                "    'driver' = 'com.mysql.jdbc.Driver',\n" +
                "    'lookup.cache.ttl' = '5min',\n" +
                "    'table-name' = 'd_idc_details',\n" +
                "    'url' = 'jdbc:mysql://tidb-lb.data.onething.internal:4000/dim?useUnicode=true&autoReconnect=true&useSSL=false&characterEncoding=UTF-8&allowMultiQueries=true',\n" +
                "    'lookup.max-retries' = '3',\n" +
                "    'username' = 'beta',\n" +
                "    'password' = 'beta@9898w',\n" +
                "    'query-sql' = 'select `idc`, `switch_name`, `name`, `isp`, `port`, `status`, `collect`, `alarm`,`bandwith`,`idc_logic`, `bandwith_link` from d_idc_details where `status` != 3'\n" +
                ");\n" +
                "create table r_i1_idc_flow_monitor_new (\n" +
                "    `stime`\ttimestamp,\n" +
                "    `groupid`\tint,\n" +
                "    `isp`\tint,\n" +
                "    `idc`\tstring,\n" +
                "    `device`\tstring,\n" +
                "    `ifname`\tstring,\n" +
                "    `iface`\tint,\n" +
                "    `flow_in`\tdouble,\n" +
                "    `flow_out`\tdouble,\n" +
                "    `is_alarm`\tint,\n" +
                "    `is_collect`\tint,\n" +
                "    `status`\tint,\n" +
                "    `build_bw`\tdouble,\n" +
                "    `idc_logic`\tstring,\n" +
                "    `bandwith_link`\tdouble\n" +
                ") with (\n" +
                "    'database' = 'ops_app_lo',\n" +
                "    'password' = 'rNXB>=dQCe3Ri-CGg*B?',\n" +
                "    'table-mode' = 'single',\n" +
                "    'sink.write-mode' = 'local',\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'sink.flush-interval' = '60s',\n" +
                "    'sink.batch-size' = '10000',\n" +
                "    'url' = 'clickhouse://ck-lb.data.onething.internal:8123',\n" +
                "    'table' = 'r_i1_idc_flow_monitor_new_test',\n" +
                "    'username' = 'spark'\n" +
                ");\n" +
                "\n" +
                "create view o_ops_miner_quality_monitor_filter as\n" +
                "select *\n" +
                "from o_ops_miner_quality_monitor\n" +
                "where `metric` in ('biz.swcollector.sinband','biz.swcollector.soutband');\n" +
                "\n" +
                "create view o_ops_miner_quality_monitor_group as\n" +
                "select\n" +
                "    split_index(split_index(tags, ',', 0), '=', 1) as idc,\n" +
                "    split_index(split_index(tags, ',', 1), '=', 1) as device,\n" +
                "    split_index(split_index(tags, ',', 2), '=', 1) as ifname,\n" +
                "    avg(case when metric='biz.swcollector.sinband' then val else null end)  as val_in,\n" +
                "    avg(case when metric='biz.swcollector.soutband' then val else null end)  as val_out,\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    window_time,\n" +
                "    proctime() as proctime\n" +
                "from table(tumble(table o_ops_miner_quality_monitor_filter, descriptor(`stime`), interval '60' second))\n" +
                "group by\n" +
                "    split_index(split_index(tags, ',', 0), '=', 1),\n" +
                "    split_index(split_index(tags, ',', 1), '=', 1),\n" +
                "    split_index(split_index(tags, ',', 2), '=', 1),\n" +
                "    window_start,window_end,window_time;\n" +
                "\n" +
                "create view o_ops_miner_quality_monitor_join as\n" +
                "select\n" +
                "    t.window_time as stime,\n" +
                "    (case when t.idc is not null then t.idc else m.idc end) as idc,\n" +
                "    (case when t.device is not null then t.device else m.switch_name end) as device,\n" +
                "    (case when t.ifname is not null then t.ifname else m.name end) as ifname,\n" +
                "    t.val_in as flow_in,\n" +
                "    t.val_out as flow_out,\n" +
                "    ifnull(m.isp,-9996) as isp,\n" +
                "    m.port as port,\n" +
                "    m.alarm as alarm,\n" +
                "    m.idc_logic as idc_logic,\n" +
                "    m.`collect` as is_collect,\n" +
                "    m.status as status,\n" +
                "    cast(m.bandwith as double) as build_bw,\n" +
                "    m.bandwith_link as bandwith_link\n" +
                "from o_ops_miner_quality_monitor_group t\n" +
                "         left join d_idc_details for system_time as of t.proctime m on t.idc=m.idc and t.device=m.switch_name and t.ifname=m.`name`;\n" +
                "\n" +
                "create view o_ops_miner_quality_monitor_join_group as\n" +
                "select\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    window_time as stime,\n" +
                "    isp,\n" +
                "    idc,\n" +
                "    device,\n" +
                "    ifname,\n" +
                "    port as iface,\n" +
                "    flow_in,\n" +
                "    flow_out,\n" +
                "    alarm as is_alarm,\n" +
                "    is_collect,\n" +
                "    status,\n" +
                "    build_bw,\n" +
                "    idc_logic,\n" +
                "    bandwith_link\n" +
                "from table(tumble(table o_ops_miner_quality_monitor_join, descriptor(`stime`), interval  '60' second))\n" +
                "group by\n" +
                "    isp,\n" +
                "    idc,\n" +
                "    device,\n" +
                "    ifname,\n" +
                "    port,\n" +
                "    flow_in,\n" +
                "    flow_out,\n" +
                "    alarm,\n" +
                "    is_collect,\n" +
                "    status,\n" +
                "    build_bw,\n" +
                "    idc_logic,\n" +
                "    bandwith_link,\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    window_time;\n" +
                "\n" +
                "insert into r_i1_idc_flow_monitor_new\n" +
                "select\n" +
                "    window_start,\n" +
                "    1 groupid,\n" +
                "    isp,\n" +
                "    idc,\n" +
                "    device,\n" +
                "    ifname,\n" +
                "    iface,\n" +
                "    flow_in,\n" +
                "    flow_out,\n" +
                "    is_alarm,\n" +
                "    is_collect,\n" +
                "    status,\n" +
                "    build_bw,\n" +
                "    idc_logic,\n" +
                "    bandwith_link\n" +
                "from o_ops_miner_quality_monitor_join_group\n" +
                "union all\n" +
                "select\n" +
                "    window_start,\n" +
                "    2 groupid,\n" +
                "    isp,\n" +
                "    '-9999' idc,\n" +
                "    '-9999' device,\n" +
                "    '-9999' ifname,\n" +
                "    -9999 iface,\n" +
                "    sum(flow_in) as flow_in,\n" +
                "    sum(flow_out) as flow_out,\n" +
                "    max(is_alarm) as is_alarm,\n" +
                "    max(is_collect) as is_collect,\n" +
                "    min(status) as status,\n" +
                "    sum(build_bw) as build_bw,\n" +
                "    '-9999' idc_logic,\n" +
                "    sum(bandwith_link) as bandwith_link\n" +
                "from o_ops_miner_quality_monitor_join_group\n" +
                "group by window_start, window_end, isp\n" +
                "union all\n" +
                "select\n" +
                "    window_start,\n" +
                "    3 groupid,\n" +
                "    isp,\n" +
                "    idc,\n" +
                "    '-9999' device,\n" +
                "    '-9999' ifname,\n" +
                "    -9999 iface,\n" +
                "    sum(flow_in) as flow_in,\n" +
                "    sum(flow_out) as flow_out,\n" +
                "    max(is_alarm) as is_alarm,\n" +
                "    max(is_collect) as is_collect,\n" +
                "    min(status) as status,\n" +
                "    sum(build_bw) as build_bw,\n" +
                "    idc_logic,\n" +
                "    sum(bandwith_link) as bandwith_link\n" +
                "from o_ops_miner_quality_monitor_join_group\n" +
                "group by window_start, window_end,isp,idc,idc_logic\n" +
                "union all\n" +
                "select\n" +
                "    window_start,\n" +
                "    4 groupid,\n" +
                "    isp,\n" +
                "    idc,\n" +
                "    device,\n" +
                "    '-9999' ifname,\n" +
                "    -9999 iface,\n" +
                "    sum(flow_in) as flow_in,\n" +
                "    sum(flow_out) as flow_out,\n" +
                "    max(is_alarm) as is_alarm,\n" +
                "    max(is_collect) as is_collect,\n" +
                "    min(status) as status,\n" +
                "    sum(build_bw) as build_bw,\n" +
                "    idc_logic,\n" +
                "    sum(bandwith_link) as bandwith_link\n" +
                "from o_ops_miner_quality_monitor_join_group\n" +
                "group by window_start, window_end,isp,idc,device,idc_logic;";




        String ss = sql + "\u0001" + "job.name=app_onecloud_sgp_bigminer_lichen\n" +
                "job.parallelism=3\n" +
                "job.max.parallelism=9\n" +
                "job.enable.checkpointing.interval=60\n" +
                "job.checkpointing.mode=exactly-once\n" +
                "job.state.backend=rocksdb\n" +
                "job.state.backend.uri=hdfs:///user/flink/calc_primary/flink-checkpoints\n" +
                "job.checkpoint.cleanup.strategy=delete\n" +
                "job.checkpoint.parallelism=1\n" +
                "job.enable.checkpoint.unaligned.enable=false\n" +
                "job.checkpoint.timeout.ms=120000\n" +
                "job.checkpoint.failure.numbers=10\n" +
                "job.restart.strategy=fixed-delay\n" +
                "job.restart.attempts=1000\n" +
                "job.restart.delay.interval=30";


        System.out.println(Base64.getEncoder().encodeToString(ss.getBytes()));
    }
}
